package study.dq.jms.ps;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 事务性消息
 *
 * @author dongqun
 * @version 1.0.0
 * @since 2021/04/20 14:53:00
 */
public class TransactionalPublisher {
    private TopicConnection connection;
    private TopicSession pubSession;
    private TopicPublisher publisher;

    public TransactionalPublisher(String topicName) {
        try {
            TopicConnectionFactory factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://127.0.0.1:61616");
            TopicConnection connect = factory.createTopicConnection();
            TopicSession pubSession = connect.createTopicSession(true, Session.SESSION_TRANSACTED);
            Topic topic = pubSession.createTopic(topicName);
            TopicPublisher publisher = pubSession.createPublisher(topic);
//            publisher.setDeliveryMode(DeliveryMode.PERSISTENT); //设置持久传递还非持久传递
            this.connection = connect;
            this.pubSession = pubSession;
            this.publisher = publisher;
            connect.start();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void sendMessage(String msg) throws JMSException {
        TextMessage message = pubSession.createTextMessage();
        message.setText(msg);
        message.setJMSDestination(publisher.getDestination());
        publisher.publish(message);
    }

    public void sendTransactionalMessage() {
        try {
            sendMessage("发送第一条消息");
            sendMessage("发送第二条消息");
            sendMessage("发送第三条消息");
            sendMessage("发送第四条消息");
            // 在事务性会话中不提交commit，消息就不会发出去
            pubSession.commit();
            // commit之后再出异常，不会再影响前面已经提交的
            int i = 1/0;
        } catch (Exception ignored) {
            try {
                pubSession.rollback();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

    public void close() throws JMSException {
        connection.close();
    }


    public static void main(String[] args) {
        TransactionalPublisher publisher = new TransactionalPublisher("topic_test");
        publisher.sendTransactionalMessage();
    }
}
